package com.zmj.tl.database.taos.kafka;

import cn.hutool.core.collection.ListUtil;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.TimeInterval;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSONArray;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.zmj.tl.database.taos.highvolume.SQLWriter;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;


/**
 * @author zhangjp
 * @version 1.0
 * @title KafkaConsum
 * @description
 * @create 2024/5/7 16:17
 */

public class KafkaConsum {
    private Statement stmt;
    private Connection conn;
    private static final Logger logger = LoggerFactory.getLogger(KafkaConsum.class);

    // 设置main方法执行时的日志输出级别

    // 需求内容：单个消费者，每秒需要处理1000条数据，每条数据的处理时间为500ms，相同accNum的数据需要保证消费的顺序

    // 测试极限情况：数据已存在大量积压，启动消费者进行消费
    // 每次拉取都达到设置的单次可以拉取的最大条数：2000条

    public static void main(String[] args) throws InterruptedException, SQLException {

        Properties props = new Properties();
        // bootstrap.servers：kafka集群地址
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.130.66:9092,192.168.130.67:9092,192.168.130.68:9092");
        // 消费者组id
        props.put("group.id", "test_consumer_group"); // 消费者组
        // key.deserializer：key的反序列化器
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // value.deserializer：value的反序列化器
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // fetch.max.bytes：一次拉取的最小可返回数据量：1Bety
        props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 100 * 1024);
        // fetch.max.bytes：一次拉取的最大数据量：50M
        props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 50 * 1024 * 1024);
        // fetch.max.wait.ms：一次拉取的最大等待时间：500ms
        props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 500);
        // max.poll.records: 一次拉取的最大条数
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 2000);
        // max.partition.fetch.bytes：一次拉取时，每个分区最大拉取数据量，默认1M
        props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 1 * 1024 * 1024);
        // auto.offset.reset：当 Kafka 中没有初始偏移量或当前偏移量在服务器中不存在 （如，数据被删除了）时，自动设置开始消费的偏移量位置，默认latest。
        // earliest：自动重置偏移量到最早的偏移量（从头开始消费）。
        // latest：默认，自动重置偏移量为最新的偏移量（从最新的接收到的数据开始消费）。
        // none：如果消费组原来的（previous）偏移量不存在，则向消费者抛异常。 anything：向消费者抛异常。
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // enable.auto.commit：是否允许自动提交offset，默认是。
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        // auto.commit.interval.ms：自动提交offset的时间间隔，默认5秒。
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 5000);
        // heartbeat.interval.ms：消费者心跳检测时间间隔，默认3秒。
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 3000);
        // session.timeout.ms：session过期时间，默认10秒。
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10000);
        // max.poll.interval.ms：一批次数据最大可以执行时间，默认5分钟。
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300000);
        // partition.assignment.strategy：分区分配策略，默认5分钟。
        props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, RangeAssignor.class.getName());

        KafkaConsumer<String, String> tKafkaConsumer = new KafkaConsumer<String, String>(props);
        tKafkaConsumer.subscribe(Arrays.asList("iot.data"));

        // 使用Executors中的CachedThreadPool，初始核心线程数为0，最大线程数为无限大，线程最大空闲时间为60秒
        // corePoolSize=0
        // maximumPoolSize=Integer.MAX_VALUE，即2147483647，基本属于无界。
        // keepAliveTime=60秒
        // 工作队列使用没有容量的 SynchronousQueue，来一个任务处理一个任务，不进行缓存。如果提交任务速度高于线程池中线程处理任务的速度，则会不断创建新线程。极端情况下会创建过多的线程，耗尽 CPU 和内存资源。
        // 可以自定义线程池进行优化
        ExecutorService executorService = Executors.newCachedThreadPool();
        SQLWriter writer = new SQLWriter(1000);
        writer.init();
        HashMap<String, List<BaseMessage>> busiDataHashMap;
        while (true) {
            TimeInterval timer = DateUtil.timer();
            logger.info("[开始]-consumer拉取数据");
            ConsumerRecords<String, String> records = tKafkaConsumer.poll(500);
            int dataCount = records.count();
            AtomicInteger tAtomicInteger = new AtomicInteger();
            logger.info("[完成]-consumer拉取数据-条数=[{}]-耗时=[{}]", dataCount, timer.intervalMs());
            // 拉取的数据条数大于0时，才进行处理操作
            timer = DateUtil.timer();
            if (dataCount > 0) {
                // 初始化hashMap：容量可以设置为拉取数据条数的1倍，或者2倍，2倍更加分散
                // 消费者参数中设置一次拉取的最大条数为2000，基本不会超过该值。
                // hashMap的hash碰撞概率较低，2000条数据，分布到4000容量的hashMap中时，基本不会出现碰撞，只有相同的key会在一起，导致整体执行时间为相同多个key顺序执行的时间
                // [线程执行完成]消费者线程：pool-1-thread-1898-已处理数据数量=3-已处理的所有客户账号=GW0032,GW0032,GW0032,
                // [线程执行完成]消费者线程：pool-1-thread-1193-已处理数据数量=2-已处理的所有客户账号=KE0055,KE0055,
                // [线程执行完成]消费者线程：pool-1-thread-1187-已处理数据数量=2-已处理的所有客户账号=0E0005,0E0005,
                int capacity = dataCount * 2;
                busiDataHashMap = new HashMap<>(capacity);
                // 将拉取的数据按客户号码分散到HashMap中
                for (ConsumerRecord<String, String> record : records) {
                    Object value = record.value();
                    String jsonStr = JSONUtil.toJsonStr(value);
                    if (ObjectUtil.isEmpty(jsonStr)) {
                        continue;
                    }
                    // logger.info("[获取]-传入报文=[{}]", jsonStr);
                    List<BaseMessage> baseMessages = new ArrayList<BaseMessage>(JSONArray.parseArray(jsonStr, BaseMessage.class));
                    // String accNum = busiDataEntity.getAccNum();

                    if (CollectionUtils.isEmpty(baseMessages)) {
                        continue;
                    }
                    if (busiDataHashMap.containsKey(baseMessages.get(0).getMetric())) {
                        busiDataHashMap.get(baseMessages.get(0).getMetric()).addAll(baseMessages);
                    } else {
                        List<BaseMessage> newList = new ArrayList<>();
                        newList.addAll(baseMessages);
                        busiDataHashMap.put(baseMessages.get(0).getMetric(), newList);
                    }
                }

                ArrayList<FutureTask<String>> tFutureTaskArrayList = new ArrayList<>(dataCount);
                // 循环hashMap，每个value开启一个线程循环处理该List中的全部数据，保证数据处理的顺序
                int num = 0;
                busiDataHashMap.forEach((k, v) -> {
                    List<BaseMessage> busiDataEntities = v;
                    String threadName = "";
                    if (busiDataEntities.size() > 0) {
                        threadName = k;
                        // 使用Callable执行同一个Key下的一组数据
                        FutureTask<String> futureTask = new FutureTask<>(new Callable<String>() {
                            @Override
                            public String call() {
                                String threadName = Thread.currentThread().getName();
                                // logger.info("[获取]-消费者线程：{}-获取到待处理数据数量：{}", threadName, busiDataEntities.size());
                                String allAccNum = "";
                                    // allAccNum = allAccNum + busiDataEntity.getAccNum() + ",";
                                List<List<BaseMessage>> partition = ListUtil.partition(busiDataEntities, 20);
                                // 模拟业务处理时间，默认500ms
                                partition.forEach(e-> {
                                    try {
                                        writer.executeSQL(sqlWrite(e));
                                    } catch (SQLException ex) {
                                        throw new RuntimeException(ex);
                                    }
                                });

                                return "消费者线程：" + threadName + "-已处理数据数量=" + busiDataEntities.size() + "-已处理的所有客户账号=" + allAccNum;
                            }
                        });

                        // 通过线程池进行任务处理
                        executorService.submit(futureTask);
                        // 将每个线程的futureTask都放入同一个ArrayList中
                        tFutureTaskArrayList.add(futureTask);
                    }
                });
                // 循环tFutureTaskArrayList，检查所有futureTask是否都已经返回，没返回的阻塞等待，等都返回后证明所有线程都执行完成，提交offset
                // 使用线程池后，线程数一直保持在2000个左右。
                for (int i = 0; i < tFutureTaskArrayList.size(); i++) {
                    try {
                        String returnStr = tFutureTaskArrayList.get(i).get();
                        logger.info("[线程执行完成]" + returnStr);
                    } catch (ExecutionException e) {
                        e.printStackTrace();
                    }
                }
            }

            // 同步提交offset
            // tKafkaConsumer.commitSync();
            // 异步提交
            tKafkaConsumer.commitAsync(new OffsetCommitCallback() {
                @Override
                public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
                    if (exception != null) {
                        logger.error("[失败]-提交offset失败！" + offsets);
                    } else {
                        logger.info("[成功]-提交offset成功！");
                    }
                }
            });

            logger.info("【完成处理数据】-条数=[{}]-耗时=[{}]", dataCount, timer.intervalMs());

        }

    }

    public static String sqlWrite(List<BaseMessage> list) {
        list.forEach(e -> e.setTName(e.getTags().get("tname").toString()));
        StringBuilder sb = new StringBuilder("INSERT INTO ");
        list.forEach(e -> sb.append(e.getTName()).append(" USING ").append(e.getMetric()).append(" TAGS ").append(packageSql(e.getTags())).append(" VALUES ").append(packageSqls(e))

        );
        return sb.toString();
    }

    public static String packageSql(Map<Object, Object> map) {
        return  "('" + map.get("tenant_id") + "','" +  map.get("device_name") + "','" + map.get("receive_time")  + "','" + map.get("device_no")  + "','" + map.get("device_type") + "','" + map.get("attribute_name")
                + "','" + map.get("message_id")  + "','" + map.get("attribute_code")  + "')";
    }
    public static String packageSqls(BaseMessage map) {
        return   "(" + map.getTimestamp() + ",'" + map.getValue() + "'," + map.getTags().get("quality") + ")  ";
    }


}
